package we.dispatcher;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import we.common.CCommonHandle;
import we.common.CDebug;
import we.config.CConstant;
import we.session.CAttachment;
import we.session.CSession;
import we.session.IoHandler;

public class CCoreServer implements Runnable
{

    private static CCoreServer _instance;
    private static final Object lock = new Object();

    public static void CloseChannel(SelectionKey key) 
    {
        try
        {
            key.channel().close();
        } catch (IOException ex)
        {
            Logger.getLogger(CCoreServer.class.getName()).log(Level.SEVERE, null, ex);
        }
        key.cancel();
        CAttachment attachment = (CAttachment) key.attachment();
        CCoreServer.instance().getHandler().SessionClosed(attachment.session);
    }
    Executor writer;
    Executor reader;
    BlockingQueue<SelectionKey> in;
    BlockingQueue<CMessage> out;
    Selector selector;
    ServerSocketChannel serverChannel;
    SelectionKey serverkey;
    IoHandler handler;

    private CCoreServer()
    {
    }

    public static CCoreServer instance()
    {
        if (_instance == null)
        {
            synchronized (lock)
            {
                if (_instance == null)
                {
                    _instance = new CCoreServer();
                }
            }
        }
        return _instance;
    }

    public void StartServer()
    {
        Executors.newSingleThreadExecutor().execute(this);
    }

    public void SetHandler(IoHandler handler)
    {
        this.handler = handler;
    }

    public IoHandler getHandler()
    {
        return handler;
    }

    private void init()
    {
        in = new LinkedBlockingQueue<SelectionKey>();
        out = new LinkedBlockingQueue<CMessage>();
        reader = Executors.newFixedThreadPool(CConstant.MAX_INQUEUE_WORKER);
        writer = Executors.newFixedThreadPool(CConstant.MAX_OUTQUEUE_WORKER);

        for (int i = 0; i < CConstant.MAX_INQUEUE_WORKER; i++)
        {
            reader.execute(new CReadHandler(in));

        }

        for (int i = 0; i < CConstant.MAX_OUTQUEUE_WORKER; i++)
        {
            writer.execute(new CWriteHandle(out));
        }

        try
        {
            selector = Selector.open();
            serverChannel = ServerSocketChannel.open();
            serverChannel.socket().bind(new java.net.InetSocketAddress(CConstant.SERVER_PORT));
            serverChannel.configureBlocking(false);
            serverkey = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e)
        {
            // TODO
            CCommonHandle.writeErrLog(e);
            System.out.println("Server is stoped");
            System.exit(1);
        }
        System.out.println("Server Old is running");
        CDebug.Trace(this, "Debug mode");
    }

    @Override
    public void run()
    {
        init();
        handler.Init();
        while (true)
        {
            try
            {
                selector.select();
                Set keys = selector.selectedKeys();
                for (Iterator i = keys.iterator(); i.hasNext();)
                {
                    SelectionKey key = (SelectionKey) i.next();
                    i.remove();
                    if (key == serverkey)
                    {
                        if (key.isAcceptable())
                        {
                            SocketChannel clientSocket = serverChannel.accept();
                            // Debug.trace("accept : " + clientSocket.socket().getInetAddress() + " : " + clientSocket.socket().getPort());
                            clientSocket.configureBlocking(false);
                            SelectionKey clientKey = clientSocket.register(selector, SelectionKey.OP_READ);
                            CAttachment attachment = new CAttachment(new CPacketReader(CConstant.MAX_MESSAGE_LENGTH), new CSession(clientKey));
                            clientKey.attach(attachment);
                            handler.SessionOpened(attachment.session);
                            
                            if (CConstant.HAND_SHAKED){
                                CMessage msg = new CMessage(clientKey, null, true);
                                out.offer(msg);
                            }
                        }
                    } else
                    {
                        in.offer(key);
                    }
                }
            } catch (IOException e)
            {
                Logger.getLogger(this.getClass().getName()).log(Level.SEVERE, "O ham run: ", e);
            }
        }
    }

    
    public void SendMsg(SelectionKey key, ByteBuffer bb)
    {
        out.offer(new CMessage(key, bb));                
    }
    
    public void sendMsgAndClone(SelectionKey key, ByteBuffer bb)
    {
        ByteBuffer tmp = bb.duplicate();
        out.offer(new CMessage(key, tmp));                
    }


    public void sendMsg(SelectionKey key, byte[] bb)
    {

        ByteBuffer tmp = ByteBuffer.allocate(bb.length);
        tmp.clear();
        tmp.put(bb);
        tmp.flip();

        out.offer(new CMessage(key, tmp));
    }
}
